package com.atguigu.day04;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SideOutputDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class Flink02_SplitStreamByOutput {
    public static void main(String[] args) throws Exception {
        //1.获取流的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        //2.从端口读取数据
        DataStreamSource<String> streamSource = env.socketTextStream("localhost", 9999);

        //TODO 定义侧输出标记
        OutputTag<String> outputTag = new OutputTag<String>("up10"){};

        OutputTag<String> outputlow5Tag = new OutputTag<String>("low5") {
        };

        //TODO 使用侧输出流来进行分流
        SingleOutputStreamOperator<String> process = streamSource.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String value, ProcessFunction<String, String>.Context ctx, Collector<String> out) throws Exception {
                //1.首先获取到侧输出
                //将大于10 的数据放到侧输出流
                if (Integer.parseInt(value) > 10) {
//                    ctx.output(new OutputTag<String>("up10") {}, value);
                    ctx.output(outputTag,value);

                } else if (Integer.parseInt(value) < 5) {
                    ctx.output(outputlow5Tag,value);
                } else {
                    //将小于等于10 的数据放到主流中
                    out.collect(value);
                }
            }
        });


        process.print("主流");

        //TODO 获取到侧输出流
//        SideOutputDataStream<String> sideOutput = process.getSideOutput(new OutputTag<String>("up10") {});
        SideOutputDataStream<String> sideOutput = process.getSideOutput(outputTag);

        process.getSideOutput(outputlow5Tag).print("小于5侧流");

        sideOutput.print("侧输出流");
        env.execute();

    }
}
